package com.yh.client;

import com.yh.config.MqttProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * MQTT 客户端组件，提供 MQTT 消息的接收和发送功能。该类使用 Spring Integration 和 MQTT 协议实现了消息的入站与出站通道的配置，
 * 便于与外部系统进行消息通信。
 * <p>
 * 核心功能包括：
 * 1. 配置消息入站通道以接收 MQTT 消息。
 * 2. 配置消息处理器以处理入站消息。
 * 3. 配置消息出站通道以发送 MQTT 消息。
 */
@Component
public class MqttClient {

    @Resource
    private MqttProperties mqttProperties;

    @Resource
    private DefaultMqttPahoClientFactory mqttClientFactory;

    /**
     * 入站通道
     * <p>
     * 用于接收外部系统传入的 MQTT 消息。消息会被路由到该通道并进一步由处理器进行处理。
     *
     * @return 入站消息通道的实例
     */
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    /**
     * 配置消息入站适配器
     * <p>
     * 从外部系统订阅指定的 MQTT Topic，接收 MQTT 消息，并将其发送到应用程序的入站通道。
     *
     * @param mqttInputChannel 用于接收消息的入站通道
     * @return 配置完成的消息生产者
     */
    @Bean
    public MessageProducer inbound(MessageChannel mqttInputChannel) {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperties.getClientId() + "_customer",
                mqttClientFactory,
                mqttProperties.getTopic());
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel);
        return adapter;
    }

    /**
     * 消息入站处理器
     * <p>
     * 处理从入站通道接收的 MQTT 消息，并执行具体的业务逻辑。
     * 例如，可以将接收到的消息保存到数据库或触发某些业务流程。
     *
     * @return 入站消息的处理器
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            // TODO 具体的业务逻辑，可以将消息保存到自己的业务
            System.out.println("Received message: " + message.getPayload());
        };
    }

    /**
     * 出站通道
     * <p>
     * 用于将应用程序内生成的 MQTT 消息路由到指定的出站适配器，从而发送到外部系统。
     *
     * @return 出站消息通道的实例
     */
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * 消息出站适配器
     * <p>
     * 处理从出站通道路由的 MQTT 消息，并通过指定的 MQTT Topic 发送到 MQTT 服务器。
     * 支持异步发送以提高性能。
     *
     * @return 配置完成的出站消息处理器
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                mqttProperties.getClientId() + "_producer",
                mqttClientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(mqttProperties.getTopic());
        return messageHandler;
    }
}
